package io.confluent.kafka.consumer.core;

import io.confluent.kafka.consumer.conf.PropertiesReader;
import io.confluent.kafka.consumer.processor.EventProcessor;
import kafka.utils.ShutdownableThread;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Collections;
import java.util.Properties;
/**
 * Created by taotao on 2017/12/18.
 */
public class ConsumerEvent extends ShutdownableThread {
    private final KafkaConsumer<String, String> consumer;
    private final String topic;

    public ConsumerEvent(String topic) {
        super("KafkaConsumer", false);
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, PropertiesReader.KAFKA_BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, PropertiesReader.KAFKA_CONSUMER_GROUP_EVENT);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        consumer = new KafkaConsumer<>(props);
        this.topic = topic;
    }

    @Override
    public void doWork() {
        consumer.subscribe(Collections.singletonList(this.topic));
        ConsumerRecords<String, String> records = consumer.poll(1000);
        EventProcessor.postData(records);
    }

    @Override
    public String name() {
        return null;
    }

    @Override
    public boolean isInterruptible() {
        return false;
    }

    private void insertIntoDb(ConsumerRecord<String, String> consumerRecord) {
        System.out.println("receive:" +
                "topic("+consumerRecord.topic() + ")," +
                "partition("+consumerRecord.partition() + ")," +
                "offset("+consumerRecord.offset() + ")," +
                "key(" + consumerRecord.key() + ")," +
                "value(" + consumerRecord.value()+")");
        System.out.println("process successful");
    }
}